package com.atguigu.day08;

import com.atguigu.util.Event;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.List;
import java.util.Map;

public class Example4 {
    public static void main(String[] args) {
        var env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // flink cep 必须使用事件时间
        var stream = env
                .addSource(new SourceFunction<Event>() {
                    @Override
                    public void run(SourceContext<Event> ctx) {
                        ctx.collectWithTimestamp(new Event("user-1", "login-fail", 1000L), 1000L);
                        ctx.collectWithTimestamp(new Event("user-1", "login-fail", 2000L), 2000L);
                        ctx.collectWithTimestamp(new Event("user-2", "login-success", 3000L), 3000L);
                        ctx.collectWithTimestamp(new Event("user-1", "login-fail", 4000L), 4000L);
                        ctx.collectWithTimestamp(new Event("user-1", "login-fail", 5000L), 5000L);
                    }

                    @Override
                    public void cancel() {

                    }
                });

        // 定义模板
        var pattern = Pattern
                .<Event>begin("first")
                .where(new SimpleCondition<>() {
                    @Override
                    public boolean filter(Event value) {
                        return value.type.equals("login-fail");
                    }
                })
                // next表示紧挨着上一个事件
                .next("second")
                .where(new SimpleCondition<>() {
                    @Override
                    public boolean filter(Event value) {
                        return value.type.equals("login-fail");
                    }
                })
                .next("third")
                .where(new SimpleCondition<>() {
                    @Override
                    public boolean filter(Event value) {
                        return value.type.equals("login-fail");
                    }
                });

        // 在数据流上检测符合模板的事件组，然后提取出来并打印
        CEP
                .pattern(stream.keyBy(r -> r.key), pattern)
                // 泛型：事件组中的事件类型，输出的类型
                .select(new PatternSelectFunction<Event, String>() {
                    @Override
                    public String select(Map<String, List<Event>> map) {
                        // map参数的结构：{
                        //    "first": [Event],
                        //    "second": [Event],
                        //    "third": [Event]
                        // }
                        var first = map.get("first").get(0);
                        var second = map.get("second").get(0);
                        var third = map.get("third").get(0);

                        return first.key + "连续三次登陆失败，登陆时间是：" + first.ts + "; " + second.ts + "; " + third.ts;
                    }
                })
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
